package com.gzhryc.shared_device.oem.micro.api.event;

import com.gzhryc.common.event.DelayHandleService;
import com.gzhryc.common.json.JsonTools;
import com.gzhryc.common.kafka.consumer.interfaces.KafkaConsumerEventListener;
import com.gzhryc.common.kafka.models.ConsumerEvent;
import com.gzhryc.common.logging.Logger;
import com.gzhryc.jdbc.JdbcSession;
import com.gzhryc.shared_device.oem.code.KafkaEventConstants;
import com.gzhryc.shared_device.oem.code.order.event.OrderCreateEvent;

import org.apache.commons.lang3.StringUtils;

import java.util.List;

public class ConfigKafkaEventListener implements KafkaConsumerEventListener {

    final static Logger log = Logger.getLogger(ConfigKafkaEventListener.class);

    String topic;

    @Override
    public void start(String topic) {
        this.topic = topic;
        log.info("开始监听主题：" + topic);
    }

    @Override
    public void receive(List<ConsumerEvent> events) {
        for (ConsumerEvent event : events) {
            String json = event.getValue();
            log.info("主题：{{0}}，组：{{1}}，区：{{2}}，获取消息，KEY={{3}}，VALUE={{4}}", event.getTopic(), event.getGroupId(),
                    event.getPartition(), event.getKey(), event.getValue());
            if (StringUtils.isNotBlank(json)) {
                try {
                    if (KafkaEventConstants.KEY_OrderCreate.equals(event.getKey())) {
                        OrderCreateEvent businessEvent = JsonTools.fromJson(json, OrderCreateEvent.class);
                        DelayHandleService.get().addEvent(businessEvent);
                    }
                }catch (Exception e) {
                    log.error("解析订单业务事件错误", e);
                } finally {
                    JdbcSession.clear();
                }
            }
        }
    }

    @Override
    public void close(String topic) {
        log.info("结束监听主题：" + topic);
    }
}
